package com.tang;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * @author tmh
 * 事物生产者
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        //第二个参数开启消息追踪
        TransactionMQProducer producer = new TransactionMQProducer("transactionProducer_producer_group");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                ///这里需要执行本地事务，如果本地事务成功的话，那么就返回成功
                //如果本地事务失败的话，那么就返回失败，则回滚事务
                //目前我这里仅仅只是模拟dmo而已，那么这里就暂时不做任何事情
                //直接返回成功，就意味着本地事务操作成功
                //直接返回失败，就以为者本地事务操作失败
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                //触发事务的检查，提供给到生产者一个检查事务是否成功提交的机会
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        List<Order> list1 = buildOrderList(1, 1, "Buy", "TG1");

        List<Order> orderList = new ArrayList<>();
        orderList.addAll(list1);

        int size = orderList.size();
        for (int i = 0; i < size; i++) {
            Order order = orderList.get(i);
            String info = order.toString();
            //这里的主题是局部有序的主题
            Message msg = new Message("Transaction-Topic", order.tag, info.getBytes());
            //setKeys这个方法就是设置每个消息的MessageKey
            msg.setKeys("Transaction_Tag");

            //直接将msg发出去
            producer.sendMessageInTransaction(msg,null);
        }
        System.out.println("发送成功");
    }

    public static List<Order> buildOrderList(int orderId, int size, String desc, String tag) {
        List<Order> resultList = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            int newOrderId = orderId + i;
            Order order = new Order();
            order.orderId = newOrderId;
            order.desc = desc;
            order.tag = tag;
            resultList.add(order);
        }
        return resultList;
    }

    public static class Order {
        /**
         * 订单id
         */
        int orderId;
        /**
         * 订单描述
         */
        String desc;

        String tag;

        @Override
        public String toString() {
            return "Order{" +
                    "orderId=" + orderId +
                    ", desc='" + desc + '\'' +
                    ", tag='" + tag + '\'' +
                    '}';
        }
    }
}